media_pp\elements\source\test/
audio.rs

1use std::{
2    f64::consts::TAU,
3    sync::Arc,
4    thread,
5    time::{Duration, Instant},
6};
7
8use crate::pp_log::{PpLog, pp_info};
9use ffmpeg_next as ffmpeg;
10use thiserror::Error as ThisError;
11
12use crate::{
13    buffer::MediaBuffer,
14    bus::{Bus, BusEvent},
15    control::{ControlReceiver, drain_control},
16    element::{Element, ElementType, Source, SourceElement, element_pp_log},
17    error::Result,
18    pad::SrcPad,
19    schedule::ActiveTimeline,
20};
21
22/// How often [`TestAudioSource::run`] wakes up to top up however many
23/// samples wall-clock time now owes — same role/value as
24/// [`crate::elements::WasapiCaptureSource`]'s own `POLL_INTERVAL`/
25/// [`crate::elements::AudioMixer`]'s `TICK_INTERVAL`.
26const TICK_INTERVAL: Duration = Duration::from_millis(20);
27
28/// Errors specific to `TestAudioSource`. Converts into the crate-wide
29/// `Error` via `?` (see [`crate::error::Error`]).
30#[derive(Debug, ThisError)]
31pub enum TestAudioSourceError {
32    #[error("TestAudioSource doesn't support seeking a generated stream")]
33    SeekUnsupported,
34}
35
36/// Construction-time options for [`TestAudioSource::new`].
37#[derive(Debug, Clone, Copy)]
38pub struct TestAudioOptions {
39    pub sample_rate: u32,
40    pub channels: u16,
41    /// The generated sine tone's frequency, in Hz. `440.0` (concert pitch
42    /// A) by default — audible, easy to recognize on a scope or by ear;
43    /// nothing else is special about the exact value.
44    pub frequency: f64,
45}
46
47impl Default for TestAudioOptions {
48    fn default() -> Self {
49        Self {
50            sample_rate: 48000,
51            channels: 2,
52            frequency: 440.0,
53        }
54    }
55}
56
57/// Generates a synthetic sine-wave tone — GStreamer's `audiotestsrc`
58/// equivalent. No real capture device involved: [`TestAudioSource::run`]
59/// fabricates however many samples wall-clock time now owes on a
60/// drift-free absolute schedule (`expected = elapsed * sample_rate`,
61/// `needed = expected - samples_emitted` — the same shape
62/// `WasapiCaptureSource::fill_silence_gap`/`AudioMixer::mix_tick` both
63/// use, not a fixed
64/// per-tick sample count, which would drift the same way a fixed-duration
65/// `thread::sleep`-only schedule would), stamps it with an increasing
66/// `pts` (one sample per tick of [`TestAudioSource::time_base`]'s units),
67/// and pushes it straight downstream — useful for exercising
68/// `AudioMixer`/an encoder/a muxer without a real microphone.
69///
70/// Always emits `Sample::F32(Packed)` — the same fixed internal format
71/// `AudioMixer` mixes in, so this can feed a `MixerHandle` input directly
72/// with nothing to resample (though `MixerInputSink` resamples regardless
73/// if fed something else instead, so this isn't load-bearing).
74///
75/// Runs until `Stop` — never reaches `Eos` on its own, same as every other
76/// live source in this crate (no sample-count limit is exposed,
77/// deliberately, mirroring a live capture source more than a file).
78pub struct TestAudioSource {
79    pp_log: PpLog,
80    name: Arc<str>,
81    pad: SrcPad,
82    sample_rate: u32,
83    channels: u16,
84    format: ffmpeg::format::Sample,
85    channel_layout: ffmpeg::ChannelLayout,
86    frequency: f64,
87    /// Cumulative sample count across every emitted frame — this
88    /// element's `pts` unit (see [`TestAudioSource::time_base`]) *and* the
89    /// sine wave's own running phase ([`TestAudioSource::generate_frame`]
90    /// divides this by `sample_rate` for `t`), so the waveform stays
91    /// phase-continuous across frame boundaries instead of restarting
92    /// from zero every tick.
93    samples_emitted: i64,
94}
95
96// SAFETY: see `AudioMixer`'s own `unsafe impl Send` docs — same
97// reasoning, `channel_layout` here is always `ChannelLayout::default`'s
98// plain native layout.
99unsafe impl Send for TestAudioSource {}
100
101impl TestAudioSource {
102    pub fn new(name: impl Into<String>, options: TestAudioOptions) -> Self {
103        let name: Arc<str> = name.into().into();
104        let pp_log = element_pp_log(ElementType::TestAudioSource, &name, None);
105        pp_info!(
106            pp_log: &pp_log,
107            "created: {}Hz, {} channel(s), {}Hz tone",
108            options.sample_rate,
109            options.channels,
110            options.frequency
111        );
112        let pad = SrcPad::new(format!("{name}_src"));
113        Self {
114            name,
115            pp_log,
116            pad,
117            sample_rate: options.sample_rate,
118            channels: options.channels,
119            format: ffmpeg::format::Sample::F32(ffmpeg::format::sample::Type::Packed),
120            channel_layout: ffmpeg::ChannelLayout::default(options.channels as i32),
121            frequency: options.frequency,
122            samples_emitted: 0,
123        }
124    }
125
126    /// The unit each emitted frame's `pts` is expressed in.
127    pub fn time_base(&self) -> ffmpeg::Rational {
128        ffmpeg::Rational::new(1, self.sample_rate as i32)
129    }
130
131    /// Fabricates the next `needed`-sample frame: the same sine tone on
132    /// every channel, phase-continuous with whatever's already been
133    /// emitted (see [`TestAudioSource::samples_emitted`]'s own docs).
134    fn generate_frame(&mut self, needed: usize) -> ffmpeg::frame::Audio {
135        let channels = self.channels as usize;
136        let mut interleaved = vec![0f32; needed * channels];
137        for (index, chunk) in interleaved.chunks_mut(channels).enumerate() {
138            let t = (self.samples_emitted + index as i64) as f64 / self.sample_rate as f64;
139            let sample = (t * self.frequency * TAU).sin() as f32;
140            chunk.fill(sample);
141        }
142
143        let mut frame = ffmpeg::frame::Audio::new(self.format, needed, self.channel_layout);
144        frame.set_rate(self.sample_rate);
145        let bytes = unsafe {
146            std::slice::from_raw_parts(
147                interleaved.as_ptr() as *const u8,
148                std::mem::size_of_val(&*interleaved),
149            )
150        };
151        // Same tight-length write `AudioMixer::mix_tick`/
152        // `WasapiCaptureSource::build_frame` both use — `data_mut(0)`'s own
153        // length is FFmpeg's own padded linesize, not necessarily exactly
154        // `bytes.len()`.
155        frame.data_mut(0)[..bytes.len()].copy_from_slice(bytes);
156        frame.set_pts(Some(self.samples_emitted));
157        self.samples_emitted += needed as i64;
158        frame
159    }
160}
161
162impl Element for TestAudioSource {
163    fn name(&self) -> Arc<str> {
164        self.name.clone()
165    }
166
167    fn element_type(&self) -> ElementType {
168        ElementType::TestAudioSource
169    }
170
171    fn pp_log(&self) -> &crate::pp_log::PpLog {
172        &self.pp_log
173    }
174
175    fn pp_log_mut(&mut self) -> &mut crate::pp_log::PpLog {
176        &mut self.pp_log
177    }
178}
179
180impl Source for TestAudioSource {
181    fn src_pads(&mut self) -> &mut [SrcPad] {
182        std::slice::from_mut(&mut self.pad)
183    }
184}
185
186impl SourceElement for TestAudioSource {
187    fn run(&mut self, control: &ControlReceiver, bus: &Bus) -> Result<()> {
188        pp_info!(self, "started");
189        let mut timeline = ActiveTimeline::new(Instant::now());
190        loop {
191            let outcome = drain_control(control, self, bus)?;
192            if outcome.stopped {
193                pp_info!(self, "stopped");
194                return Ok(());
195            }
196            timeline.account_pause(outcome.paused_for);
197            thread::sleep(TICK_INTERVAL);
198
199            let expected =
200                (timeline.elapsed(Instant::now()).as_secs_f64() * self.sample_rate as f64) as i64;
201            let needed = (expected - self.samples_emitted).max(0) as usize;
202            if needed == 0 {
203                continue;
204            }
205            let frame = self.generate_frame(needed);
206            // A downstream failure drops just this one frame — same
207            // "report, don't die" contract every other source in this
208            // crate gives its own push.
209            if let Err(error) = self.pad.push(MediaBuffer::Audio(Arc::new(frame))) {
210                bus.post(
211                    &self.pp_log,
212                    BusEvent::Error {
213                        element_type: ElementType::TestAudioSource,
214                        name: self.name.clone(),
215                        error,
216                    },
217                );
218            }
219        }
220    }
221
222    fn seek(&mut self, _target: Duration) -> Result<Duration> {
223        Err(TestAudioSourceError::SeekUnsupported.into())
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use std::sync::Mutex;
230
231    use crate::pp_log::PpLog;
232
233    use super::*;
234    use crate::{control::ControlMsg, element::Sink, pipeline::Pipeline};
235
236    /// Captures every frame's `(format, rate, channels, pts, first_sample)`
237    /// it sees, in order.
238    struct RecordingSink {
239        pp_log: PpLog,
240        #[allow(clippy::type_complexity)]
241        seen: Arc<Mutex<Vec<(ffmpeg::format::Sample, u32, u16, Option<i64>, f32)>>>,
242    }
243
244    impl Element for RecordingSink {
245        fn name(&self) -> Arc<str> {
246            "recorder".into()
247        }
248        fn element_type(&self) -> ElementType {
249            ElementType::Other
250        }
251        fn pp_log(&self) -> &PpLog {
252            &self.pp_log
253        }
254        fn pp_log_mut(&mut self) -> &mut PpLog {
255            &mut self.pp_log
256        }
257    }
258
259    impl Sink for RecordingSink {
260        fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
261            if let MediaBuffer::Audio(frame) = buf
262                && frame.samples() > 0
263            {
264                self.seen.lock().unwrap().push((
265                    frame.format(),
266                    frame.rate(),
267                    frame.channel_layout().channels() as u16,
268                    frame.pts(),
269                    frame.plane::<f32>(0)[0],
270                ));
271            }
272            Ok(())
273        }
274        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
275            Ok(())
276        }
277    }
278
279    #[test]
280    fn generates_f32_frames_with_increasing_pts_and_a_bounded_tone() {
281        let seen = Arc::new(Mutex::new(Vec::new()));
282        let sink = RecordingSink {
283            seen: seen.clone(),
284            pp_log: element_pp_log(ElementType::Other, "recorder", None),
285        };
286        let source = TestAudioSource::new(
287            "test-audio",
288            TestAudioOptions {
289                sample_rate: 48000,
290                channels: 2,
291                frequency: 440.0,
292            },
293        );
294
295        let pipeline = Pipeline::new("test", source, |source, ctx| {
296            let branch = ctx.branch().to(Box::new(sink))?;
297            ctx.attach(source, 0, branch)?;
298            Ok(())
299        })
300        .expect("test pipeline wiring must succeed");
301
302        pipeline.run();
303        // Long enough to observe several ticks at the 20ms `TICK_INTERVAL`.
304        std::thread::sleep(Duration::from_millis(200));
305        pipeline.stop();
306        pipeline.bus().log_events();
307
308        let frames = seen.lock().unwrap();
309        assert!(!frames.is_empty(), "expected at least one generated frame");
310        for &(format, rate, channels, _, sample) in frames.iter() {
311            assert_eq!(
312                format,
313                ffmpeg::format::Sample::F32(ffmpeg::format::sample::Type::Packed)
314            );
315            assert_eq!((rate, channels), (48000, 2));
316            assert!(
317                (-1.0..=1.0).contains(&sample),
318                "expected a bounded sine sample, got {sample}"
319            );
320        }
321        for window in frames.windows(2) {
322            assert!(
323                window[1].3 > window[0].3,
324                "expected pts to strictly increase frame over frame, got {:?} then {:?}",
325                window[0].3,
326                window[1].3
327            );
328        }
329    }
330
331    #[test]
332    fn seek_is_explicitly_unsupported() {
333        let mut source = TestAudioSource::new("test-audio", TestAudioOptions::default());
334        assert!(source.seek(Duration::from_secs(1)).is_err());
335    }
336
337    /// Regression test for the pause/resume timing bug: `start.elapsed()`
338    /// keeps advancing while [`Pipeline::pause`] blocks this source's own
339    /// loop inside `drain_control`. Without subtracting the accumulated
340    /// `ControlOutcome::paused_for` back out, `Resume` would find the
341    /// whole pause suddenly counted as owed samples and emit one wildly
342    /// oversized frame to cover it, instead of resuming its steady
343    /// per-tick sample count — each frame's `pts` is a running sample
344    /// count, so a healthy run never has two consecutive frames whose
345    /// `pts` gap is anywhere near a whole pause's worth of samples.
346    #[test]
347    fn resuming_after_a_pause_does_not_dump_a_burst_of_samples() {
348        let seen = Arc::new(Mutex::new(Vec::new()));
349        let sink = RecordingSink {
350            seen: seen.clone(),
351            pp_log: element_pp_log(ElementType::Other, "recorder", None),
352        };
353        let source = TestAudioSource::new(
354            "test-audio",
355            TestAudioOptions {
356                sample_rate: 48000,
357                channels: 2,
358                frequency: 440.0,
359            },
360        );
361
362        let pipeline = Pipeline::new("pause-resume-test", source, |source, ctx| {
363            let branch = ctx.branch().to(Box::new(sink))?;
364            ctx.attach(source, 0, branch)?;
365            Ok(())
366        })
367        .expect("test pipeline wiring must succeed");
368
369        pipeline.run();
370        thread::sleep(Duration::from_millis(60));
371        pipeline.pause();
372        thread::sleep(Duration::from_millis(400));
373        pipeline.resume();
374        thread::sleep(Duration::from_millis(100));
375        pipeline.stop();
376        pipeline.bus().log_events();
377
378        let frames = seen.lock().unwrap();
379        let pts: Vec<i64> = frames.iter().filter_map(|&(_, _, _, pts, _)| pts).collect();
380        assert!(
381            pts.len() >= 2,
382            "expected multiple frames spanning the pause/resume, got {}",
383            pts.len()
384        );
385        for window in pts.windows(2) {
386            let gap = window[1] - window[0];
387            // A healthy tick's worth of samples at 48kHz/20ms is ~960; a
388            // 400ms pause treated as owed catch-up would show up as a
389            // ~19200-sample gap. 12000 (250ms) sits comfortably between
390            // the two.
391            assert!(
392                gap < 12_000,
393                "expected steady per-tick sample counts across resume, not a single burst \
394                 frame covering the whole pause: consecutive pts gap was {gap} samples \
395                 ({:.0}ms) — full pts sequence: {pts:?}",
396                gap as f64 / 48.0
397            );
398        }
399    }
400}
media_pp\elements\source\test/audio.rs

media_pp\elements\source\test/
audio.rs
